package rabbitmq

import (
	"errors"

	"git.100tal.com/wangxiao_dahai_backend/arch-nova/middleware"
	"github.com/assembla/cony"
	"github.com/streadway/amqp"
)

// 单消费者
func (mq *RabbitMq) NewConsumer(e *Exchange, q *Queue, key string, opts ...cony.ConsumerOpt) error {
	if e.Name == "" || e.Type == "" {
		return errors.New("exchange name or kind error")
	}
	if !checkExchangeType(e.Type) {
		return errors.New("exchange type error")
	}

	que := &cony.Queue{}
	que.Name = q.Name
	que.Durable = q.Durable
	que.AutoDelete = q.AutoDelete
	que.Exclusive = q.Exclusive
	que.Args = q.Args

	exc := cony.Exchange{}
	exc.Name = e.Name
	exc.Kind = e.Type
	exc.Durable = e.Durable
	exc.AutoDelete = e.AutoDelete
	exc.Args = e.Args

	bnd := cony.Binding{
		Queue:    que,
		Exchange: exc,
		Key:      key,
	}

	// Declare and register a consumer
	cns := cony.NewConsumer(
		que,
		opts...,
	)
	mq.cli.Consume(cns)

	mq.cli.Declare([]cony.Declaration{
		cony.DeclareQueue(que),
		cony.DeclareExchange(exc),
		cony.DeclareBinding(bnd),
	})

	// Client loop sends out declarations(exchanges, queues, bindings
	// etc) to the AMQP server. It also handles reconnecting.
	go func() {
		defer func() {
			if err := recover(); err != nil {
				middleware.LogError(nil, "[RecoveryPanic] [RabbitMq Consumer] "+err.(string))
			}
		}()

		for mq.cli.Loop() {
			select {
			case msg := <-cns.Deliveries():
				// fmt.Println("Received body: " + string(msg.Body))
				// mq.Message <- string(msg.Body)
				// 由业务控制ack等
				mq.Delivery <- msg

				// If when we built the consumer we didn't use
				// the "cony.AutoAck()" option this is where we'd
				// have to call the "amqp.Deliveries" methods "Ack",
				// "Nack", "Reject"
				//
				// msg.Ack(false)
				// msg.Nack(false, false)
				// msg.Reject(false)
			case err := <-cns.Errors():
				e := errors.New("mq_consumer_client error: " + err.Error())
				mq.Error <- e

			case err := <-mq.cli.Errors():
				e := errors.New("mq_consumer_client error: " + err.Error())
				mq.Error <- e

			}
		}
	}()

	return nil
}

// 单消费者简化版接口 direct
// 无须配置exchange queue等
// 默认持久化
func (mq *RabbitMq) NewConsumerDirect(exchangeName, queueName, routeKey string, opts ...cony.ConsumerOpt) error {
	e := NewExchange(
		ExchangeName(exchangeName),
		ExchangeType(EXCHANGE_TYPE_DIRECT),
		ExchangeDurable(true),
	)
	q := NewQueue(
		QueueName(queueName),
		QueueDurable(true),
	)
	err := mq.NewConsumer(e, q, routeKey, opts...)
	if err != nil {
		return err
	}

	return nil
}

// 单消费者简化版接口 fanout
// 无须配置exchange queue等
// 默认持久化
func (mq *RabbitMq) NewConsumerFanout(exchangeName, queueName, routeKey string, opts ...cony.ConsumerOpt) error {
	e := NewExchange(
		ExchangeName(exchangeName),
		ExchangeType(EXCHANGE_TYPE_FANOUT),
		ExchangeDurable(true),
	)
	q := NewQueue(
		QueueName(queueName),
		QueueDurable(true),
	)
	err := mq.NewConsumer(e, q, routeKey, opts...)
	if err != nil {
		return err
	}

	return nil
}

// 多消费者包装
type ConsumerWrapper struct {
	cns *cony.Consumer
	e   *cony.Exchange
	q   *cony.Queue
	k   string
}

func (mq *MqWrapper) NewConsumerWrapper(e *Exchange, q *Queue, key string, opts ...cony.ConsumerOpt) (*ConsumerWrapper, error) {
	if e.Name == "" || e.Type == "" {
		return nil, errors.New("exchange name or type empty")
	}
	if !checkExchangeType(e.Type) {
		return nil, errors.New("exchange type error")
	}

	que := &cony.Queue{}
	if q != nil {
		que.Name = q.Name
		que.Durable = q.Durable
		que.AutoDelete = q.AutoDelete
		que.Exclusive = q.Exclusive
		que.Args = q.Args
	}

	exc := cony.Exchange{}
	exc.Name = e.Name
	exc.Kind = e.Type
	exc.Durable = e.Durable
	exc.AutoDelete = e.AutoDelete
	exc.Args = e.Args

	bnd := cony.Binding{
		Queue:    que,
		Exchange: exc,
		Key:      key,
	}

	// Declare and register a consumer
	cns := cony.NewConsumer(
		que,
		opts...,
	)
	mq.cli.Consume(cns)

	mq.cli.Declare([]cony.Declaration{
		cony.DeclareQueue(que),
		cony.DeclareExchange(exc),
		cony.DeclareBinding(bnd),
	})

	return &ConsumerWrapper{
		cns: cns,
		e:   &exc,
		q:   que,
		k:   key,
	}, nil
}

// Deliveries return deliveries shipped to this consumer
// this channel never closed, even on disconnects
func (c *ConsumerWrapper) Deliveries() <-chan amqp.Delivery {
	return c.cns.Deliveries()
}

// Errors returns channel with AMQP channel level errors
func (c *ConsumerWrapper) Errors() <-chan error {
	return c.cns.Errors()
}

func (c *ConsumerWrapper) GetExchangeName() string {
	if c.e != nil {
		return c.e.Name
	}

	return ""
}

func (c *ConsumerWrapper) GetQueueName() string {
	if c.q != nil {
		return c.q.Name
	}

	return ""
}

func (c *ConsumerWrapper) GetRoutingKey() string {
	return c.k
}
